package com.sjc.lesson02.state.keyedState.function;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class ContainsValueFunction extends RichFlatMapFunction<Tuple2<Long,Long>, Tuple2<Long,String>> {

    private AggregatingState<Long,String> totalStr;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        AggregatingStateDescriptor<Long, String, String> descriptor =
                new AggregatingStateDescriptor<>("totalStr", // 状态的名字
                    new AggregateFunction<Long, String, String>() {
                        @Override
                        public String createAccumulator() {
                            //  创建一个新的累加器
                            return "Contains: ";
                        }

                        @Override
                        public String add(Long value, String accumulator) {
                            // 将新的数据与累加器进行聚合，返回一个新的累加器
                            if ("Contains: ".equals(accumulator)) {
                                return accumulator + value;
                            }
                            return accumulator + "add " + value;
                        }

                        @Override
                        public String getResult(String accumulator) {
                            // 从累加器中计算最终结果并返回
                            return accumulator;
                        }

                        @Override
                        public String merge(String a, String b) {
                            // 合并两个累加器并返回结果
                            return a + " and " + b;
                        }
                    }, String.class);// 状态存储的数据类型

        totalStr = getRuntimeContext().getAggregatingState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> element, Collector<Tuple2<Long, String>> collector) throws Exception {
        totalStr.add(element.f1);
        collector.collect(Tuple2.of(element.f0, totalStr.get()));
    }
}
